#include "threadPool.h"

int epoll_add(int epfd, int fd) {
  struct epoll_event event;
  event.events = EPOLLIN;
  event.data.fd = fd;
  epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event);
  return 0;
}

int epoll_del(int epfd, int fd) {
  epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
  return 0;
}

int threadPool_init(threadPool *pool, int work_num) {
  threadArr_init(&pool->t_arr, work_num);
  queue_init(&pool->queue);
  mysql_client_init(&pool->mysql, work_num);
  pthread_mutex_init(&pool->mutex, NULL);
  pthread_cond_init(&pool->cond, NULL);
  pool->user_map = user_table_create();
  user_table_init(pool->user_map);
  pool->is_close = 0;
  
  return 0;
}

int threadPool_make_worker(threadPool *pool) {

  pthread_mutex_lock(&pool->mutex);

  for (int i = 0; i < pool->t_arr.work_num; ++i) {
   pthread_create(&pool->t_arr.tid[i], NULL, worker, pool);
   int_table_put(pool->t_arr.index, pool->t_arr.tid[i], i);
  }

  pthread_mutex_unlock(&pool->mutex);

  return 0;
}

void *worker(void *arg) {

  threadPool *pool = (threadPool *)arg;

  pthread_mutex_lock(&pool->mutex);

  MYSQL *conn = mysql_init(NULL);
  sql_config s;
  conf_mysql(&s);
  MYSQL *ret = mysql_real_connect(conn, s.host, s.user_name, s.pass_word,
                                  s.data_base, s.port, NULL, 0);
  if (!ret) {
    fprintf(stderr, "mysql_real_connect:%s\n", mysql_error(conn));
    return NULL;
  }

  int tid_index;
  pthread_t tid = pthread_self();
  
  int ret_int_table_find =
      int_table_find(pool->t_arr.index, tid, &tid_index);
  if (ret_int_table_find != 0) {
    fprintf(stderr, "int_table_find_error in threadPool.c:worker.\n");
    return NULL;
  }
  pool->mysql[tid_index] = conn;

  pthread_mutex_unlock(&pool->mutex);
  
  while (1) {
    
    pthread_mutex_lock(&pool->mutex);

    while (pool->is_close == 0 && pool->queue.size == 0) {
      pthread_cond_wait(&pool->cond, &pool->mutex);
    }
    if (pool->is_close == 1) {
      pthread_mutex_unlock(&pool->mutex);
      pthread_exit(NULL);
    }
    Task t = pool->queue.front->task;
    deque(&pool->queue);

    pthread_mutex_unlock(&pool->mutex);

    if (t.func) {
      t.func(t.arg, conn);
    }
    free_para((Para*)t.arg);
  }
}

void free_para(Para *para) {
  if (para == NULL) {
    return;
  }
  if (para->path != NULL) {
    for (int i = 0; para->path[i] != NULL; ++i) {
      free(para->path[i]);
    }
    free(para->path);
  }
  free(para);
}
